package reactor;

import org.junit.Test;
import reactor.core.publisher.Flux;
import reactor.core.publisher.SynchronousSink;
import util.Util;

import java.time.Duration;
import java.util.concurrent.CountDownLatch;
import java.util.function.BiFunction;
import java.util.function.Supplier;

public class TimeoutTest {

    private static CountDownLatch countDownLatch = new CountDownLatch(1);

    @Test
    public void testTimeout1() throws InterruptedException {
        Flux.just(((Supplier<Integer>) () -> {
            Util.sleep(400L);
            return 1;
        }).get())
                .timeout(Duration.ofMillis(300), Flux.just(2))
                .subscribe(
                        (x) -> Util.print("消费了元素" + x),
                        Throwable::printStackTrace,
                        () -> countDownLatch.countDown());
        countDownLatch.await();
    }

    @Test
    public void testTimeout2() throws InterruptedException {
        Flux.generate(() -> 1, (BiFunction<Integer, SynchronousSink<Integer>, Integer>) (state, sink) -> {
            Util.sleep(400);
            sink.next(state);
            sink.complete();
            return state;
        }).timeout(Duration.ofMillis(300), Flux.just(2))
                .subscribe(
                        (x) -> Util.print("消费了元素" + x),
                        Throwable::printStackTrace,
                        () -> countDownLatch.countDown());
        countDownLatch.await();
    }
}
